#include "Common.h"
#include "UDPPort.h"
#include "Poco/Observer.h"
#include "Poco/Net/NetException.h"

using Poco::Observer;
using Poco::Net::NetException;
using Poco::Net::ConnectionResetException;
using Poco::Logger;

namespace cm_base
{

    UDPPort::UDPPort(SocketReactor& reactor,ByteBufferPool* pool)
        :m_reactor(reactor),m_buffer_pool(pool), m_stop(false),m_listener(NULL)
    {
    }

    UDPPort::~UDPPort()
    {

    }

    int UDPPort::setup(const IPAddress& ip)
    {
        SocketAddress sa(ip, 0);
        return setup(sa);
    }

    int UDPPort::setup(const SocketAddress& sa)
    {
        try
        {
            m_socket.bind(sa);
        }
        catch(NetException& e)
        {
            goto fail;
        }

        m_hostAddress = m_socket.address();

        m_reactor.addEventHandler(m_socket, 
            Observer<UDPPort, ReadableNotification>(*this, &UDPPort::onReceive) );

        m_thread.start( *this );

        return CM_RES_SUCCESS;

    fail:
        return CM_RES_FAIL;
    }

    int UDPPort::teardown()
    {
        m_stop = true;

        return CM_RES_SUCCESS;
    }

    SocketAddress UDPPort::getHostAddress()
    {
        return m_socket.address();
    }

    void UDPPort::registerListener(UDPPortListener* p_listener)
    {
        m_listener = p_listener;
    }

    int UDPPort::send(ByteBuffer* p_data,const SocketAddress& sa)
    {
        if( p_data ){
            SocketAddress* p_sa = new SocketAddress(sa);
            p_data->setData((void*)p_sa);
            p_data->setType(EVENT_TO_SEND_DATA);
            m_queue.enqueueNotification(p_data);
        }
        return CM_RES_SUCCESS;
    }

    void UDPPort::onReceive(ReadableNotification * sn)
    {
        sn->release();

        ByteBuffer* p_data = m_buffer_pool->get(0);
        if( p_data ){
            p_data->increaseRef();
            SocketAddress* p_sa = new SocketAddress;
            int recv = 0;
            try
            {
                recv = m_socket.receiveFrom( (void*)p_data->Data(), p_data->Capacity(), *p_sa);
            }
            catch(NetException& e) 
            {
                poco_debug(g_log_base, e.message());
            }
            p_data->setLength(recv);
            p_data->setData(p_sa);
            p_data->setType(EVENT_RECVED_DATA);

            m_queue.enqueueNotification( p_data );
        }
    }

    void UDPPort::run()
    {
        BaseNotification* p_event = NULL;

        while( !m_stop )
        {
            p_event = (BaseNotification* )m_queue.waitDequeueNotification();

            switch( p_event->getType() )
            {
            case EVENT_RECVED_DATA:
                {
                    ByteBuffer* p_data = (ByteBuffer*)p_event;
                    if( m_listener ){
                        m_listener->onReceive(this, p_data );
                    }else{
                        SocketAddress* p_sa = (SocketAddress*)p_data->getData();
                        delete p_sa;
                        p_data->decreaseRef();
                    }
                }
                break;

            case EVENT_TO_SEND_DATA:
                {
                    ByteBuffer* p_data = (ByteBuffer*)p_event;

                    SocketAddress* p_sa = (SocketAddress*)p_data->getData();
                    //poco_debug( g_log_base, Logger::format("UDPPort Send:$0", p_sa->toString()) );

                    try
                    {
                        m_socket.sendTo(p_data->Data(), p_data->Length(), *p_sa);
                    }
                    catch(NetException& e)
                    {
                        poco_debug( g_log_base, e.message() );
                    }

                    delete p_sa;
                    p_data->decreaseRef();
                }
                break;

            default:
                {
                    p_event->release();
                }
                break;
            }
        }
    }

}//cm_base